package com.gjy.kafka.sp.config;

import com.gjy.kafka.sp.entity.MessageEntity;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.util.HashMap;
import java.util.Map;

import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
import static org.apache.kafka.clients.producer.ProducerConfig.*;

/**
 * Kafka配置类
 * 配置Kafka主题、生产者、消费者等
 *
 * @author gjy
 * @version 1.0
 * @since 2025-09-07 10:35:28
 */
@Configuration
public class KafkaConfig {

    /**
     * 创建普通消息主题
     * 3个分区，1个副本
     *
     * @return 普通消息主题
     */
    @Bean
    public NewTopic normalTopic() {
        // 参数：主题名称、分区数、副本数
        return new NewTopic(KafkaConstant.NORMAL_TOPIC, 3, (short) 1);
    }

    /**
     * 创建分区消息主题
     * 5个分区，1个副本
     *
     * @return 分区消息主题
     */
    @Bean
    public NewTopic partitionTopic() {
        return new NewTopic(KafkaConstant.PARTITION_TOPIC, 5, (short) 1);
    }

    /**
     * 创建事务消息主题
     * 3个分区，1个副本
     *
     * @return 事务消息主题
     */
    @Bean
    public NewTopic transactionalTopic() {
        return new NewTopic(KafkaConstant.TRANSACTIONAL_TOPIC, 3, (short) 1);
    }

    /**
     * 创建死信主题
     * 1个分区，1个副本
     *
     * @return 死信主题
     */
    @Bean
    public NewTopic deadLetterTopic() {
        return new NewTopic(KafkaConstant.DEAD_LETTER_TOPIC, 1, (short) 1);
    }

    /**
     * 配置事务生产者工厂
     *
     * @return 事务生产者工厂
     */
    @Bean
    public ProducerFactory<String, MessageEntity> transactionalProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(BOOTSTRAP_SERVERS_CONFIG, "192.168.253.142:9092,192.168.253.142:9093,192.168.253.142:9094");
        configProps.put(KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        configProps.put(VALUE_SERIALIZER_CLASS_CONFIG, org.springframework.kafka.support.serializer.JsonSerializer.class);
        configProps.put(ACKS_CONFIG, "all");
        configProps.put(RETRIES_CONFIG, 3);
        configProps.put(BATCH_SIZE_CONFIG, 16384);
        configProps.put(BUFFER_MEMORY_CONFIG, 33554432);
        // 配置事务ID前缀
        configProps.put(TRANSACTIONAL_ID_CONFIG, KafkaConstant.TRANSACTION_ID_PREFIX);

        DefaultKafkaProducerFactory<String, MessageEntity> factory = new DefaultKafkaProducerFactory<>(configProps);
        // 开启事务支持
        factory.transactionCapable();
        return factory;
    }

    /**
     * 配置事务Kafka模板
     *
     * @return 事务Kafka模板
     */
    @Bean(name = "transactionalKafkaTemplate")
    public KafkaTemplate<String, MessageEntity> transactionalKafkaTemplate() {
        return new KafkaTemplate<>(transactionalProducerFactory());
    }

    /**
     * 配置Kafka事务管理器
     *
     * @return Kafka事务管理器
     */
    @Bean
    public KafkaTransactionManager<String, MessageEntity> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(transactionalProducerFactory());
    }

}
